import asyncio
import logging
import random
import uuid
from typing import Optional

from redis.asyncio import Redis
from redis.exceptions import RedisError

logger = logging.getLogger(__name__)


class LockError(Exception):
    """分布式锁异常"""

    pass


class DistributedLock:
    """
    基于Redis的分布式锁实现，支持自动续期和异常处理
    """

    def __init__(
        self,
        redis_client: Redis,
        lock_key: str,
        expire_seconds: int = 5,
        retry_interval_ms: int = 200,
        retry_times: int = 3,
        blocking: bool = True,
        max_renewal_times: int = 0,  # 最大续期次数，0表示不限制
        auto_renewal: bool = True,  # 是否自动续期
        retry_delay_range: tuple = (0.1, 0.3),
    ):
        """
        初始化分布式锁

        Args:
            redis_client: Redis客户端
            lock_key: 锁的键名
            expire_seconds: 锁的过期时间（秒）
            retry_interval_ms: 重试间隔（毫秒）
            retry_times: 重试次数
            blocking: 是否阻塞等待
            max_renewal_times: 最大续期次数，0表示不限制
            auto_renewal: 是否自动续期
            retry_delay_range: 重试延迟的随机范围(最小值, 最大值)，单位秒
        """
        if not lock_key or not isinstance(lock_key, str):
            raise ValueError("锁键名不能为空且必须是字符串类型")

        if not redis_client:
            raise ValueError("Redis客户端不能为空")

        self.redis = redis_client

        self.lock_key = f"lock:{lock_key}"
        self.lock_value = str(uuid.uuid4())
        self.expire_seconds = max(expire_seconds, 5)  # 确保最小过期时间
        self.retry_interval = retry_interval_ms / 1000
        self.retry_times = retry_times if blocking else 1
        self.acquired = False
        self._watchdog_task: Optional[asyncio.Task] = None
        self.max_renewal_times = max_renewal_times
        self.auto_renewal = auto_renewal
        self._renewal_count = 0
        self._should_exit = False
        self.retry_delay_range = retry_delay_range

    async def __aenter__(self):
        success = await self.acquire()
        if not success:
            raise LockError(f"无法获取锁 {self.lock_key}")
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        try:
            await self.release()
        except Exception as e:
            logger.error(f"释放锁 {self.lock_key} 时发生错误: {e}")
            raise

    async def acquire(self) -> bool:
        """
        获取锁，支持重试制

        Returns:
            bool: 是否成功获取锁

        Raises:
            LockError: Redis操作异常
        """
        retry_count = 0
        while retry_count < self.retry_times:
            try:
                if await self._acquire_lock():
                    self.acquired = True
                    # 只有在需要自动续期时才启动看门狗
                    if self.auto_renewal:
                        self._watchdog_task = asyncio.create_task(self._watchdog(), name=f"watchdog-{self.lock_key}")
                    # logger.debug(f"成功获取锁 {self.lock_key}")
                    return True

                retry_count += 1
                if retry_count < self.retry_times:
                    # 使用随机延迟避免锁竞争
                    delay = random.uniform(self.retry_delay_range[0], self.retry_delay_range[1])
                    # logger.debug(f"获取锁 {self.lock_key} 失败，{delay:.2f}秒后重试 (第{retry_count}次)")
                    await asyncio.sleep(delay)

            except RedisError as e:
                logger.error(f"获取锁 {self.lock_key} 时发生Redis错误: {e}")
                raise LockError(f"Redis错误: {e}")

        logger.warning(f"获取锁 {self.lock_key} 失败，已重试 {retry_count} 次")
        return False

    async def release(self):
        """
        释放锁，确保只能由加锁者释放

        Raises:
            LockError: Redis操作异常
        """
        # 标记看门狗需要退出
        self._should_exit = True

        if self._watchdog_task:
            self._watchdog_task.cancel()
            try:
                await self._watchdog_task
            except asyncio.CancelledError:
                pass
            except Exception as e:
                logger.error(f"取消看门狗任务时发生错误: {e}")

        if self.acquired:
            try:
                lua_script = """
                if redis.call('get', KEYS[1]) == ARGV[1] then
                    return redis.call('del', KEYS[1])
                else
                    return 0
                end
                """
                result = await self.redis.eval(lua_script, 1, self.lock_key, self.lock_value)

            except RedisError as e:
                logger.error(f"释放锁 {self.lock_key} 时发生Redis错误: {e}")
                raise LockError(f"Redis错误: {e}")
            finally:
                self.acquired = False
                self._renewal_count = 0

    async def _acquire_lock(self) -> bool:
        """
        尝试获取锁

        Returns:
            bool: 是否成功获取锁

        Raises:
            RedisError: Redis操作异常
        """
        success = await self.redis.set(self.lock_key, self.lock_value, nx=True, ex=self.expire_seconds)
        return bool(success)

    async def _watchdog(self):
        """
        看门狗任务，负责自动续期锁

        - 只在需要时续期
        - 支持最大续期次数限制
        - 可以通过标记优雅退出
        """
        try:
            while not self._should_exit:
                # 检查是否达到最大续期次数
                if self.max_renewal_times > 0 and self._renewal_count >= self.max_renewal_times:
                    logger.warning(f"锁 {self.lock_key} 已达到最大续期次数 {self.max_renewal_times}")
                    break

                # 提前1/3过期时间续期
                await asyncio.sleep(self.expire_seconds / 3)

                # 如果已标记退出，则不再续期
                if self._should_exit:
                    break

                try:
                    # 使用Lua脚本确保原子性
                    lua_script = """
                    if redis.call('get', KEYS[1]) == ARGV[1] then
                        return redis.call('expire', KEYS[1], ARGV[2])
                    else
                        return 0
                    end
                    """
                    success = await self.redis.eval(lua_script, 1, self.lock_key, self.lock_value, self.expire_seconds)

                    if success:
                        self._renewal_count += 1
                        logger.debug(
                            f"锁 {self.lock_key} 续期成功 "
                            f"(第 {self._renewal_count}/{self.max_renewal_times if self.max_renewal_times > 0 else '∞'} 次)"
                        )
                    else:
                        logger.warning(f"锁 {self.lock_key} 续期失败，可能已过期或被释放")
                        break

                except RedisError as e:
                    logger.error(f"锁 {self.lock_key} 续期时发生Redis错误: {e}")
                    break

        except Exception as e:
            logger.error(f"看门狗任务异常 {self.lock_key}: {e}")
